package com.neusoft.bd.myflink.step05;

import com.neusoft.bd.myflink.entity.WaterSensor;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;

import java.beans.Encoder;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.List;

public class SinkFile {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<WaterSensor> data = Arrays.asList(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_1", 2, 2),
                new WaterSensor("sensor_2", 2, 2),
                new WaterSensor("sensor_3", 3, 3));
        DataStreamSource<WaterSensor> ds = env.fromCollection(data);
        env.setParallelism(2);
        // 必须开启checkpoint，否则一直都是 .inprogress
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        SingleOutputStreamOperator<String> map = ds.map(new RichMapFunction<WaterSensor, String>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                RuntimeContext runtimeContext = getRuntimeContext();
                System.out.println("NumberOfParallelSubtasks->" + runtimeContext.getNumberOfParallelSubtasks()
                        + runtimeContext.getIndexOfThisSubtask() + " 的任务的生命周期开始");

            }

            @Override
            public String map(WaterSensor ws) throws Exception {
                return ws.toString();
            }

            @Override
            public void close() throws Exception {
                super.close();
            }
        });

//        FileSink fileSink =  FileSink.<String>forRowFormat(new Path("output") ,
//                new SimpleStringEncoder<>("utf8"))
//                .withOutputFileConfig(
//                        OutputFileConfig.builder()
//                                .withPartPrefix("neusoft-")
//                        .withPartSuffix(".log")
//                        .build()
//                )
//                .withBucketAssigner( new DateTimeBucketAssigner<>("yyyy-MM-dd HH", ZoneId.systemDefault()));
//               // .withRollingPolicy()


        ds.print();
        env.execute();
    }
}
